/* Copyright (c) 2008, dZettaApart.
   Licensed under the BSD License.
   See COPYING for details. */

#include <boost/bind.hpp>
#include <boost/asio.hpp>
#include <boost/thread.hpp>

#include <deque>
#include <iostream>

#include "memcacheasio_p.h"
#include "memcacheasio_request.h"
#include "memcacheasio_exception.h"
#include "memcacheasio_connection.h"
#include "memcacheasio_connectionspool.h"

using namespace memcacheasio;
using namespace std;
using namespace boost;
using namespace boost::asio;
 
namespace memcacheasio
{
	/** **************************************************************************
	 * Memcacheasio_p::Memcacheasio_p
	 * 
	 ****************************************************************************/
	Memcacheasio_p::Memcacheasio_p ( ) :
			_io_service(),
			_work ( new io_service::work ( _io_service ) )
	{
		_pool.reset ( new ConnectionsPool ( _io_service ) );
	}

	/** **************************************************************************
	 * Memcacheasio_p::Memcacheasio_p
	 * 
	 ****************************************************************************/
	Memcacheasio_p::~Memcacheasio_p()
	{
	}

	/** **************************************************************************
	 * Memcacheasio::Create
	 * нельзя в конструкторе делать  shared_from_this()
	 ****************************************************************************/
	void Memcacheasio_p::init ( const std::string& host, const std::string& port, int count_connection, int count_thread, int tryCount, int tryTimeout, int limitConnection )
	{
		_pool->init ( host, port, count_connection, tryCount, tryTimeout, limitConnection );

		// And when we have eternity - run thread
		for ( int i = 0; i < count_thread; i ++ )
		{
			boost::shared_ptr<boost::thread> thread ( new boost::thread ( boost::bind ( &Memcacheasio_p::run, shared_from_this() ) ) );
			_threadList.push_back ( thread );
		}
	}

	/** **************************************************************************
	 * Memcacheas::run
	 * 
	 ****************************************************************************/
	std::size_t Memcacheasio_p::run()
	{
		std::size_t ret = 0;

		for ( ;; )
		{
			try
			{
				ret = _io_service.run();
				break;
			}
			catch ( std::exception &e )
			{
			}
		}
		return ret;
	}

	/** **************************************************************************
	 * Memcacheasio::stop
	 * 
	 ****************************************************************************/
	void Memcacheasio_p::stop()
	{
		_pool->stop();

		_work.reset();

		for ( int i = 0; i < _threadList.size() ; i++ )
		{
			_threadList[i]->join();
		}
	}

	/** **************************************************************************
	 * Memcacheasio::execute
	 * 
	 * Add request to Memcacheasio obejct for processing
	 ****************************************************************************/
	void Memcacheasio_p::execute ( Request::Ptr request, boost::function1<void, Request::Ptr > cb )
	{
		_pool->getConnection ( boost::bind ( &Memcacheasio_p::processRequest, shared_from_this(), request, cb, _1, _2 ) );
	}

	/** **************************************************************************
	 * Memcacheasio::processRequest
	 * 
	 ****************************************************************************/
	void Memcacheasio_p::processRequest ( Request::Ptr request, boost::function1<void, Request::Ptr > cb, Connection::Ptr c_connection, const boost::system::error_code& err )
	{
		if ( c_connection && !err )
		{
			c_connection->processRequest ( request,
			                              boost::bind ( &Memcacheasio_p::completeRequest,
			                                            shared_from_this(), cb, _1, _2 ),
			                              boost::bind ( &ConnectionsPool::queueConnection,
			                                            _pool, _1, _2 ) );
		}
		else
		{
			if ( c_connection )
			{
				_pool->queueConnection ( err, c_connection );
			}
			if ( err )
			{
				_io_service.post ( boost::bind ( &Memcacheasio_p::completeRequest, shared_from_this(), cb, err, request ) );
			}
			else
			{
				boost::system::error_code not_connected_error ( boost::asio::error::not_connected );
				_io_service.post ( boost::bind ( &Memcacheasio_p::completeRequest, shared_from_this(), cb, not_connected_error, request ) );
			}
		}
	}

	/** ***************************************************************************
	 *  Memcacheasio::completeRequest
	 *
	 *****************************************************************************/
	void Memcacheasio_p::completeRequest ( boost::function1<void, Request::Ptr > cb, const boost::system::error_code& err, Request::Ptr request )
	{
		if ( err )
		{
			cb ( request );
		}
		else
		{
			cb ( request );
		}
	}
} //namespace memcacheasio

